package cn.doitedu.flink.demos;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/2/24
 **/
@Slf4j
public class StatefulTest {
    public static void main(String[] args) throws Exception {

        Configuration config = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<String> source = env.addSource(SourcesTest.createSocketSource("localhost", 9999));

        source.keyBy( s -> s.split(",")[0]).map(new ListStateRichMapFunction()).print();

        env.execute();
    }

    /**
     * value state
     */
    static class ValueStateMapFunction extends RichMapFunction<String,String> {
        ValueState accu = null;
        @Override
        public void open(Configuration parameters) throws Exception {
            ValueStateDescriptor desc = new ValueStateDescriptor("test",String.class);
            accu = getRuntimeContext().getState(desc);

        }

        @Override
        public String map(String value) throws Exception {

            if(accu.value() == null) accu.update("START-");
            String res = accu.value() + value;

            accu.update(value.toUpperCase()+"-");

            return res;
        }
    }


    /**
     * list state
     */
    static class ListStateRichMapFunction extends RichMapFunction<String,String>{
        ListState<String> listState;
        @Override
        public void open(Configuration parameters) throws Exception {

            ListStateDescriptor<String> listDesc = new ListStateDescriptor<>("list", String.class);
            listState = getRuntimeContext().getListState(listDesc);

        }

        @Override
        public String map(String value) throws Exception {

            String ele = value.split(",")[1];
            listState.add(ele);

            if(ele.equals("5")) listState.clear();

            return StringUtils.join(listState.get(),",");
        }
    }

}
